package catalina.connection;

import catalina.startup.LifeCycle;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class NioEndPoint implements LifeCycle {

    private int port;
    private int acceptCount = 100;

    private int minSpareThreads = 10;

    private int maxThreads = 200;
    private Executor executor = null;

    private Poller[] pollers;

    private AtomicInteger pollerRotater = new AtomicInteger(0);

    protected Acceptor[] acceptors;

    private int pollerThreadCount = Math.min(2, Runtime.getRuntime().availableProcessors());

    private Connector connector;

    /**
     * Acceptor thread count.
     */
    protected int acceptorThreadCount = 1;

    public static final int OP_REGISTER = 0x100; //register interest op

    public void setPollerThreadCount(int pollerThreadCount) {
        this.pollerThreadCount = pollerThreadCount;
    }

    public void setAcceptorThreadCount(int acceptorThreadCount) {
        this.acceptorThreadCount = acceptorThreadCount;
    }

    public int getAcceptorThreadCount() {
        return acceptorThreadCount;
    }

    public int getPollerThreadCount() {
        return pollerThreadCount;
    }

    public int getMinSpareThreads() {
        return minSpareThreads;
    }

    public void setMinSpareThreads(int minSpareThreads) {
        this.minSpareThreads = minSpareThreads;
    }


    public int getMaxThreads() {
        return maxThreads;
    }

    public void setMaxThreads(int maxThreads) {
        this.maxThreads = maxThreads;
    }

    public void setExecutor(Executor executor) {
        this.executor = executor;
    }

    public Executor getExecutor() {
        return executor;
    }

    public int getPort() {
        return port;
    }

    public void setAcceptCount(int acceptCount) {
        if (acceptCount > 0) {
            this.acceptCount = acceptCount;
        }
    }

    public int getAcceptCount() {
        return acceptCount;
    }


    public void setPort(int port) {
        this.port = port;
    }

    private ServerSocketChannel serverSock;

    public NioEndPoint(Connector connector) {
        this.connector = connector;
    }

    public void bind() throws Exception {
        serverSock = ServerSocketChannel.open();
        InetSocketAddress addr = new InetSocketAddress(getPort());
        serverSock.socket().bind(addr, getAcceptCount());
        serverSock.configureBlocking(true);
    }

    @Override
    public void init() {
        try {
            bind();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void start() {
        // Create worker collection
        if (getExecutor() == null) {
            createExecutor();
        }
        // Start poller threads
        pollers = new Poller[getPollerThreadCount()];
        for (int i = 0; i < pollers.length; i++) {
            pollers[i] = new Poller();
            Thread pollerThread = new Thread(pollers[i], "TP" + "-ClientPoller-" + i);
            pollerThread.setPriority(5);
            pollerThread.setDaemon(true);
            pollerThread.start();
        }
        startAcceptorThreads();
    }

    @Override
    public void stop() {

    }

    @Override
    public void destroy() {

    }

    protected final void startAcceptorThreads() {
        int count = getAcceptorThreadCount();
        acceptors = new Acceptor[count];
        for (int i = 0; i < count; i++) {
            acceptors[i] = new Acceptor();
            String threadName = "TP" + "-Acceptor-" + i;
            acceptors[i].setThreadName(threadName);
            Thread t = new Thread(acceptors[i], threadName);
            t.setPriority(5);
            t.setDaemon(true);
            t.start();
        }
    }

    public void createExecutor() {
        BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
        executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), 60, TimeUnit.SECONDS, taskQueue, threadFactory, handler);
    }

    public class Poller implements Runnable {
        public Selector getSelector() {
            return selector;
        }

        public void setSelector(Selector selector) {
            this.selector = selector;
        }

        private Selector selector;
        private final BlockingQueue<PollerEvent> events =
                new LinkedBlockingQueue<>(128);
        private long selectorTimeout = 1000;

        public Poller() {
            try {
                this.selector = Selector.open();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        public void register(PollerEvent event) {
            try {
                events.put(event);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        public void register(NioChannel nioChannel) {
            nioChannel.setPoller(this);
            NioSocketWrapper snw = new NioSocketWrapper(nioChannel);
            snw.setInterestOps(SelectionKey.OP_READ);
            snw.setPoller(this);
            PollerEvent pollerEvent = new PollerEvent(nioChannel, snw, OP_REGISTER);
            events.offer(pollerEvent);
        }

        public void add(NioChannel nioChannel, int interestOps) {
            nioChannel.setPoller(this);
            NioSocketWrapper snw = new NioSocketWrapper(nioChannel);
            snw.setInterestOps(interestOps);
            snw.setPoller(this);
            PollerEvent pollerEvent = new PollerEvent(nioChannel, snw, interestOps);
            events.offer(pollerEvent);
        }


        @Override
        public void run() {
            while (true) {
                try {
                    boolean hasEvents = events();
                    int keyCount = selector.select(selectorTimeout);
                    if (keyCount == 0) {
                        hasEvents = events();
                    }
                    Iterator<SelectionKey> iterator = hasEvents ? selector.selectedKeys().iterator() : null;
                    while (iterator != null && iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        NioSocketWrapper nioSocketWrapper = (NioSocketWrapper) key.attachment();
                        processKey(key, nioSocketWrapper);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        public void processKey(SelectionKey sk, NioSocketWrapper attachment) {
            if (sk.isValid() && attachment != null) {
                if (sk.isReadable() || sk.isWritable()) {
                    unreg(sk, attachment, sk.readyOps());
                    boolean closeSocket = false;
                    // Read goes before write
                    if (sk.isReadable()) {
                        if (!processSocket(attachment, SocketEvent.OPEN_READ, true)) {
                            closeSocket = true;
                        }
                    }
                    if (!closeSocket && sk.isWritable()) {
                        if (!processSocket(attachment, SocketEvent.OPEN_WRITE, true)) {
                            closeSocket = true;
                        }
                    }
                }
            }
        }

        protected void unreg(SelectionKey sk, NioSocketWrapper attachment, int readyOps) {
            //this is a must, so that we don't have multiple threads messing with the socket
            reg(sk, attachment, sk.interestOps() & (~readyOps));
        }

        protected void reg(SelectionKey sk, NioSocketWrapper attachment, int intops) {
            sk.interestOps(intops);
            attachment.interestOps(intops);
        }

        public boolean events() {
            PollerEvent pollerEvent = events.poll();
            if (pollerEvent != null) {
                pollerEvent.run();
                return true;
            }
            return false;
        }

        public boolean processSocket(NioSocketWrapper socketWrapper,
                                     SocketEvent event, boolean dispatch) {
            SocketProcessor socketProcessor = new SocketProcessor(socketWrapper, event);
            Executor executor = getExecutor();
            if (dispatch && executor != null) {
                executor.execute(socketProcessor);
            } else {
                socketProcessor.run();
            }
            return true;
        }
    }

    public class Acceptor implements Runnable {
        private String threadName;

        public void setThreadName(final String threadName) {
            this.threadName = threadName;
        }

        @Override
        public void run() {
            while (true) {
                try {
                    SocketChannel socketChannel = serverSock.accept();
                    System.out.println(socketChannel);
                    setSocketOptions(socketChannel);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        public void setSocketOptions(SocketChannel socketChannel) {
            try {
                socketChannel.configureBlocking(false);
                NioChannel nioChannel = new NioChannel(socketChannel);
                getPoller0().register(nioChannel);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static class NioSocketWrapper {
        public NioChannel getSocket() {
            return socket;
        }

        public void setSocket(NioChannel socket) {
            this.socket = socket;
        }

        private NioChannel socket;

        public Poller getPoller() {
            return poller;
        }

        public void setPoller(Poller poller) {
            this.poller = poller;
        }

        public int getInterestOps() {
            return interestOps;
        }

        public void setInterestOps(int interestOps) {
            this.interestOps = interestOps;
        }

        private Poller poller = null;
        private int interestOps = 0;

        public NioSocketWrapper(NioChannel socket) {
            this.socket = socket;
        }

        public void interestOps(int ops) {
            this.interestOps = ops;
        }
    }

    /**
     * 轮询方式获取pollers数组中的poller
     *
     * @return
     */
    public Poller getPoller0() {
        int idx = Math.abs(pollerRotater.incrementAndGet()) % pollers.length;
        return pollers[idx];
    }

    public static class PollerEvent {
        private NioChannel nioChannel;
        private int interestOps; // 感兴趣的事件
        private NioSocketWrapper socketWrapper;

        public PollerEvent(NioChannel nc, NioSocketWrapper w, int intOps) {
            this.nioChannel = nc;
            this.socketWrapper = w;
            this.interestOps = intOps;
        }

        public void run() {
            if (interestOps == OP_REGISTER) {
                try {
                    nioChannel.getIOChannel().register(
                            nioChannel.getPoller().getSelector(), SelectionKey.OP_READ, socketWrapper);
                } catch (Exception x) {
                    x.printStackTrace();
                }
            } else {
                final SelectionKey key = nioChannel.getIOChannel().keyFor(nioChannel.getPoller().getSelector());
                if (key != null) {
                    final NioSocketWrapper socketWrapper = (NioSocketWrapper) key.attachment();
                    if (socketWrapper != null) {
                        //we are registering the key to start with, reset the fairness counter.
                        int ops = key.interestOps() | interestOps;
                        socketWrapper.interestOps(ops);
                        key.interestOps(ops);
                    }
                }
            }
        }

        @Override
        public String toString() {
            return "PollerEvent{" +
                    "nioChannel=" + nioChannel +
                    ", interestOps=" + interestOps +
                    ", socketWrapper=" + socketWrapper +
                    '}';
        }
    }

    public class SocketProcessor implements Runnable {
        protected NioSocketWrapper socketWrapper;
        protected SocketEvent event;

        public SocketProcessor(NioSocketWrapper socketWrapper, SocketEvent event) {
            this.socketWrapper = socketWrapper;
            this.event = event;
        }

        @Override
        public void run() {
            NioChannel socket = socketWrapper.getSocket();
            SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());  // 用作关闭以及取消key的作用
            int state = 0;
            if (event == null) {
                state = process(socketWrapper, SocketEvent.OPEN_READ);
            } else {
                state = process(socketWrapper, event);
            }
        }

        public int process(NioSocketWrapper socketWrapper, SocketEvent socketEvent) {
            NioChannel socket = socketWrapper.getSocket();
            if (SocketEvent.OPEN_READ.equals(socketEvent)) {
                doRead(socket); // 处理读请求
                socket.getPoller().add(socket, SelectionKey.OP_WRITE); // 请求处理完成之后，添加写事件
            } else if (SocketEvent.OPEN_WRITE.equals(socketEvent)) {
                doWrite(socket);
            }
            return 1;
        }

        private void doRead(NioChannel socket) {
            ByteBuffer readBuffer = socket.getReadBuffer();
            SocketChannel ioChannel = socket.getIOChannel();
            StringBuilder stringBuilder = new StringBuilder();
            int len = 0;
            while (true) {
                try {
                    if (!((len = ioChannel.read(readBuffer)) > 0)) break;
                } catch (IOException e) {
                    e.printStackTrace();
                }
                readBuffer.flip();
                // 将数据读取出来
                stringBuilder.append(new String(readBuffer.array(), 0, len));
                readBuffer.clear();
            }
            if(stringBuilder.toString().equals("")){
                return ;
            }
            Request request = new Request(stringBuilder.toString());
            Response response = new Response();
            connector.getStandardService().getContainer().service(request, response);
            byte[] bytes = response.getData().getBytes();
            ByteBuffer byteBuffer = ByteBuffer.allocate(bytes.length);
            byteBuffer.put(bytes);
            socket.setWriteBuffer(byteBuffer);
        }

        private void doWrite(NioChannel nioChannel) {
            ByteBuffer writeBuffer = nioChannel.getWriteBuffer();
            writeBuffer.flip();
            SocketChannel ioChannel = nioChannel.getIOChannel();
            try {
                ioChannel.write(writeBuffer);
            } catch (IOException e) {
                e.printStackTrace();
            }
            writeBuffer.clear();
        }
    }
}
